package prictice.hbase.state;

import org.apache.storm.task.IMetricsContext;
import org.apache.storm.trident.state.State;
import org.apache.storm.trident.state.StateFactory;
import org.apache.storm.trident.state.StateType;
import org.apache.storm.trident.state.map.*;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @Package prictice.hbase.state
 * @Description:
 * @Author elwyn
 * @Date 2017/9/21 23:02
 * @Email elonyong@163.com
 */
public class HBaseAggregateFactory implements StateFactory {
    private static final long serialVersionUID = 9006453857538332984L;
    private StateType stateType;
    private TridentConfig tridentConfig;

    @Override
    public State makeState(Map map, IMetricsContext iMetricsContext, int i, int i1) {
        HbaseAggregateState hbaseAggregateState = new HbaseAggregateState(tridentConfig);
        CachedMap c = new CachedMap(hbaseAggregateState, tridentConfig.getStateCacheSize());
        MapState mapState=null;
        if (stateType == StateType.NON_TRANSACTIONAL) {
            mapState = NonTransactionalMap.build(c);
        } else if (stateType == StateType.OPAQUE) {
            mapState = OpaqueMap.build(c);
        } else if (stateType == StateType.TRANSACTIONAL) {
            mapState = TransactionalMap.build(c);
        }
        return new SnapshottableMap<>(mapState, new Values("$GLOBAL$"));
    }

    public HBaseAggregateFactory(TridentConfig tridentConfig, StateType stateType) {
        this.tridentConfig = tridentConfig;
        this.stateType = stateType;
        if (tridentConfig.getSerializer() == null) {
            tridentConfig.setSerializer(TridentConfig.DEFAULT_SERIALIZES.get(stateType));
        }
    }
}
